package com.zhj.rabbitmq.confirms;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @program: rabbimmq-demo
 * @description: 消费者确认模式
 * @author: zhengjibin
 * @create: 2020-11-19 10:58
 * @Version 1.0
 **/
public class ReceiveConfirmMessage {

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        //配置rabbitmq的链接信息
        factory.setHost("122.112.198.94");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");
        Connection connection = null;// 定义rabbitmq的网络连接 ，一个连接
        Channel channel = null;//定义一个链接中的通道
        connection = factory.newConnection();//通过工厂获得连接
        channel = connection.createChannel();//通过连接获得通道

        String queueName = "listener_queue";

        //开启事务
        channel.txSelect();
        // 参数2:为自动确认消息
        channel.basicConsume(queueName,false,"",new DefaultConsumer(channel){
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //super.handleDelivery(consumerTag, envelope, properties, body);
                boolean redeliver = envelope.isRedeliver();
                System.out.println("消息是否被接收过，true表示接收过，false表示未接收过：："+redeliver);//用于防重复接收
                //System.out.println(10/0);//这段代码会报错，但是消息却被就收了。消息队列的消息少了。但是却什么都没有输出，这个时候就消息丢失了
                if (!redeliver){
                    String message = new String(body,"UTF-8");
                    System.out.println(message);
                    Long tag = envelope.getDeliveryTag();
                    System.out.println("消息编号："+envelope.getDeliveryTag());
                    //获得 DefaultConsumer 对象中的 Channel(通道)
                    Channel c = this.getChannel();
                    //确认消息已接收
                    //参数2：是否确认多个消息，true会确认小于等于当前编号的所有消息
                    c.basicAck(tag,true);

                    //如果消息不是自动确认,并且开启了事务。这个时候必须要提交事务。否则消息队列中的消息被消费了不会被删除，还存在
                    c.txCommit();
                }else {
                    //可以通过程序或数据库中去查询判断这个消息是否处理
                }

            }
        });
    }
}
